package com.bd08.flink.demo

import org.apache.commons.lang3.time.DateUtils
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingProcessingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object TestJoinAndCoGroup extends App {

  case class OrderLogEvent1(orderId:Long,amount:Double,timeStamp:Long)
  case class OrderLogEvent2(orderId:Long,itemId:Long,timeStamp:Long)
  case class OrderResultEvent(orderId:Long,amount:Double,itemId:Long)
  val env = StreamExecutionEnvironment.getExecutionEnvironment

  val leftOrderStream = env.fromCollection(List(
    OrderLogEvent1(1L, 22.1, DateUtils.parseDate("2020-04-29 13:01","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent1(2L, 22.2, DateUtils.parseDate("2020-04-29 13:03","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent1(4L, 22.4, DateUtils.parseDate("2020-04-29 13:04","yyyy-MM-dd HH:mm").getTime),

    OrderLogEvent1(4L, 22.3, DateUtils.parseDate("2020-04-29 13:05","yyyy-MM-dd HH:mm").getTime),

    OrderLogEvent1(5L, 22.5, DateUtils.parseDate("2020-04-29 13:07","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent1(6L, 22.6, DateUtils.parseDate("2020-04-29 13:09","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent1(2L, 22.2, DateUtils.parseDate("2020-04-29 13:10","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent1(4L, 22.4, DateUtils.parseDate("2020-04-29 13:11","yyyy-MM-dd HH:mm").getTime)
  ))
    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderLogEvent1](Time.seconds(5)) {
      override def extractTimestamp(element: OrderLogEvent1): Long = element.timeStamp
    })


  val rightOrderStream = env.fromCollection(List(
    OrderLogEvent2(1L, 121, DateUtils.parseDate("2020-04-29 13:01","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(2L, 122, DateUtils.parseDate("2020-04-29 13:03","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(3L, 123, DateUtils.parseDate("2020-04-29 13:04","yyyy-MM-dd HH:mm").getTime),

    OrderLogEvent2(4L, 125, DateUtils.parseDate("2020-04-29 13:06","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(4L, 124, DateUtils.parseDate("2020-04-29 13:05","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(5L, 125, DateUtils.parseDate("2020-04-29 13:07","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(7L, 126, DateUtils.parseDate("2020-04-29 13:09","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(3L, 123, DateUtils.parseDate("2020-04-29 13:04","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(2L, 126, DateUtils.parseDate("2020-04-29 13:10","yyyy-MM-dd HH:mm").getTime),
    OrderLogEvent2(4L, 126, DateUtils.parseDate("2020-04-29 13:11","yyyy-MM-dd HH:mm").getTime)
  ))
    .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderLogEvent2](Time.seconds(5)) {
      override def extractTimestamp(element: OrderLogEvent2): Long = element.timeStamp
    })



  /*leftOrderStream.join(rightOrderStream)
    .where(_.orderId).equalTo(_.orderId)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .apply((o1,o2,out:Collector[OrderResultEvent]) =>{
      out.collect(OrderResultEvent(o1.orderId,o1.amount,o2.itemId))
    }).print()*/
  leftOrderStream.coGroup(rightOrderStream)
    .where(_.orderId).equalTo(_.orderId)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .apply((it1:Iterator[OrderLogEvent1],it2:Iterator[OrderLogEvent2],out:Collector[String]) => {

      var rs = ""


      it1.foreach(o => {
        rs = rs + "|it1"+o
      })

      it2.foreach(o => {
        rs = rs + "|it2"+o
      })


      out.collect(rs)
    })
    .print()



 /* val ds1=env.fromElements((1,1),(2,2),(1,3),(2,4))
  val ds2=env.fromElements((1,1),(2,6),(1,7),(2,8))
  // 指定两个相连的数据集
  ds1.join(ds2)
    //指定连接条件
    .where(e => {
      println("where..........")
      e._1
    }).equalTo(e => {
    println("equalTo .....")
    e._1
  })
    //指定聚合范围
    //.window(TumblingProcessingTimeWindows.of(Time.seconds(1)))

    // 编聚合逻辑
    .apply((input1,input2,out:Collector[((Int,Int),(Int,Int))]) =>{
      println(".........")
      out.collect((input1,input2))
    }).print()*/


  env.execute("join and cogroup")




}
